package drds.binlog;

import drds.binlog.event.*;
import drds.binlog.event.mariadb.*;
import drds.binlog.parse.driver.packets.GtidSet;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.io.IOException;
import java.util.BitSet;

/**
 * Implements a binary-log decoder.
 * <p>
 * <pre>
 * LogDecoder decoder = new LogDecoder();
 * decoder.handle(...);
 *
 * LogEvent event;
 * do
 * {
 *     event = decoder.decode(bytes, context);
 *
 *     // process log event.
 * }
 * while (event != null);
 * // no more events in bytes.
 * </pre>
 *
 * @author <a href="mailto:changyuan.lh@taobao.com">Changyuan.lh</a>
 * @version 1.0
 */
public final class LogDecoder
{

    protected static final Log logger = LogFactory.getLog(LogDecoder.class);

    protected final BitSet handleSet = new BitSet(LogEvent.ENUM_END_EVENT);

    public LogDecoder()
    {
    }

    public LogDecoder(final int fromIndex, final int toIndex)
    {
        handleSet.set(fromIndex, toIndex);
    }

    /**
     * Deserialize an event from bytes.
     *
     * @return <code>UknownLogEvent</code> if event type is unknown or skipped.
     */
    public static LogEvent decode(LogByteBlock logByteBlock, LogHeader logHeader, LogContext logContext) throws IOException
    {
        FormatDescriptionLogEvent formatDescriptionLogEvent = logContext.getFormatDescriptionLogEvent();
        LogPosition logPosition = logContext.getLogPosition();

        int checksumAlg = LogEvent.BINLOG_CHECKSUM_ALG_UNDEF;
        if (logHeader.getType() != LogEvent.FORMAT_DESCRIPTION_EVENT)
        {
            checksumAlg = formatDescriptionLogEvent.logHeader.getChecksumAlg();
        } else
        {
            // 如果是format事件自己，也需要处理checksum
            checksumAlg = logHeader.getChecksumAlg();
        }

        if (checksumAlg != LogEvent.BINLOG_CHECKSUM_ALG_OFF && checksumAlg != LogEvent.BINLOG_CHECKSUM_ALG_UNDEF)
        {
            // remove checksum bytes
            logByteBlock.limit(logHeader.getEventLength() - LogEvent.BINLOG_CHECKSUM_LEN);
        }
        GtidSet gtidSet = logContext.getGtidSet();
        GtidLogEvent gtidLogEvent = logContext.getGtidLogEvent();
        switch (logHeader.getType())
        {
            case LogEvent.QUERY_EVENT:
            {
                QueryLogEvent queryLogEvent = new QueryLogEvent(logHeader, logByteBlock, formatDescriptionLogEvent);
                /* updating index$originIndex in context */
                logPosition.position = logHeader.getLogPosition();
                logHeader.putGtid(logContext.getGtidSet(), gtidLogEvent);
                return queryLogEvent;
            }
            case LogEvent.XID_EVENT:
            {
                XidLogEvent xidLogEvent = new XidLogEvent(logHeader, logByteBlock, formatDescriptionLogEvent);
                /* updating index$originIndex in context */
                logPosition.position = logHeader.getLogPosition();
                logHeader.putGtid(logContext.getGtidSet(), gtidLogEvent);
                return xidLogEvent;
            }
            case LogEvent.TABLE_MAP_EVENT:
            {
                TableMapLogEvent tableMapLogEvent = new TableMapLogEvent(logHeader, logByteBlock, formatDescriptionLogEvent);
                /* updating index$originIndex in context */
                logPosition.position = logHeader.getLogPosition();
                logContext.putTableMapLogEvent(tableMapLogEvent);
                return tableMapLogEvent;
            }
            case LogEvent.WRITE_ROWS_EVENT_V1:
            {
                RowsLogEvent rowsLogEvent = new WriteRowsLogEvent(logHeader, logByteBlock, formatDescriptionLogEvent);
                /* updating index$originIndex in context */
                logPosition.position = logHeader.getLogPosition();
                rowsLogEvent.fillTableMapLogEvent(logContext);
                return rowsLogEvent;
            }
            case LogEvent.UPDATE_ROWS_EVENT_V1:
            {
                RowsLogEvent rowsLogEvent = new UpdateRowsLogEvent(logHeader, logByteBlock, formatDescriptionLogEvent);
                /* updating index$originIndex in context */
                logPosition.position = logHeader.getLogPosition();
                rowsLogEvent.fillTableMapLogEvent(logContext);
                return rowsLogEvent;
            }
            case LogEvent.DELETE_ROWS_EVENT_V1:
            {
                RowsLogEvent rowsLogEvent = new DeleteRowsLogEvent(logHeader, logByteBlock, formatDescriptionLogEvent);
                /* updating index$originIndex in context */
                logPosition.position = logHeader.getLogPosition();
                rowsLogEvent.fillTableMapLogEvent(logContext);
                return rowsLogEvent;
            }
            case LogEvent.ROTATE_EVENT:
            {
                RotateLogEvent rotateLogEvent = new RotateLogEvent(logHeader, logByteBlock, formatDescriptionLogEvent);
                /* updating index$originIndex in context */
                logPosition = new LogPosition(rotateLogEvent.getFilename(), rotateLogEvent.getPosition());
                logContext.setLogPosition(logPosition);
                return rotateLogEvent;
            }
            case LogEvent.LOAD_EVENT:
            case LogEvent.NEW_LOAD_EVENT:
            {
                LoadLogEvent loadLogEvent = new LoadLogEvent(logHeader, logByteBlock, formatDescriptionLogEvent);
                /* updating index$originIndex in context */
                logPosition.position = logHeader.getLogPosition();
                return loadLogEvent;
            }
            case LogEvent.SLAVE_EVENT: /* can never happen (unused event) */
            {
                if (logger.isWarnEnabled()) logger.warn("Skipping unsupported SLAVE_EVENT from: "
                        + logContext.getLogPosition());
                break;
            }
            case LogEvent.CREATE_FILE_EVENT:
            {
                CreateFileLogEvent createFileLogEvent = new CreateFileLogEvent(logHeader, logByteBlock, formatDescriptionLogEvent);
                /* updating index$originIndex in context */
                logPosition.position = logHeader.getLogPosition();
                return createFileLogEvent;
            }
            case LogEvent.APPEND_BLOCK_EVENT:
            {
                AppendBlockLogEvent appendBlockLogEvent = new AppendBlockLogEvent(logHeader, logByteBlock, formatDescriptionLogEvent);
                /* updating index$originIndex in context */
                logPosition.position = logHeader.getLogPosition();
                return appendBlockLogEvent;
            }
            case LogEvent.DELETE_FILE_EVENT:
            {
                DeleteFileLogEvent deleteFileLogEvent = new DeleteFileLogEvent(logHeader, logByteBlock, formatDescriptionLogEvent);
                /* updating index$originIndex in context */
                logPosition.position = logHeader.getLogPosition();
                return deleteFileLogEvent;
            }
            case LogEvent.EXEC_LOAD_EVENT:
            {
                ExecuteLoadLogEvent executeLoadLogEvent = new ExecuteLoadLogEvent(logHeader, logByteBlock, formatDescriptionLogEvent);
                /* updating index$originIndex in context */
                logPosition.position = logHeader.getLogPosition();
                return executeLoadLogEvent;
            }
            case LogEvent.START_EVENT_V3:
            {
                /* This is sent only by MySQL <=4.x */
                StartLogEventV3 startLogEventV3 = new StartLogEventV3(logHeader, logByteBlock, formatDescriptionLogEvent);
                /* updating index$originIndex in context */
                logPosition.position = logHeader.getLogPosition();
                return startLogEventV3;
            }
            case LogEvent.STOP_EVENT:
            {
                StopLogEvent event = new StopLogEvent(logHeader, logByteBlock, formatDescriptionLogEvent);
                /* updating index$originIndex in context */
                logPosition.position = logHeader.getLogPosition();
                return event;
            }
            case LogEvent.INTVAR_EVENT:
            {
                IntvarLogEvent intvarLogEvent = new IntvarLogEvent(logHeader, logByteBlock, formatDescriptionLogEvent);
                /* updating index$originIndex in context */
                logPosition.position = logHeader.getLogPosition();
                return intvarLogEvent;
            }
            case LogEvent.RAND_EVENT:
            {
                RandLogEvent randLogEvent = new RandLogEvent(logHeader, logByteBlock, formatDescriptionLogEvent);
                /* updating index$originIndex in context */
                logPosition.position = logHeader.getLogPosition();
                logHeader.putGtid(logContext.getGtidSet(), gtidLogEvent);
                return randLogEvent;
            }
            case LogEvent.USER_VAR_EVENT:
            {
                UserVarLogEvent userVarLogEvent = new UserVarLogEvent(logHeader, logByteBlock, formatDescriptionLogEvent);
                /* updating index$originIndex in context */
                logPosition.position = logHeader.getLogPosition();
                logHeader.putGtid(logContext.getGtidSet(), gtidLogEvent);
                return userVarLogEvent;
            }
            case LogEvent.FORMAT_DESCRIPTION_EVENT:
            {
                formatDescriptionLogEvent = new FormatDescriptionLogEvent(logHeader, logByteBlock, formatDescriptionLogEvent);
                logContext.setFormatDescriptionLogEvent(formatDescriptionLogEvent);
                return formatDescriptionLogEvent;
            }
            case LogEvent.PRE_GA_WRITE_ROWS_EVENT:
            {
                if (logger.isWarnEnabled()) logger.warn("Skipping unsupported PRE_GA_WRITE_ROWS_EVENT from: "
                        + logContext.getLogPosition());
                // ev = new Write_rows_log_event_old(buf, event_len,
                // description_event);
                break;
            }
            case LogEvent.PRE_GA_UPDATE_ROWS_EVENT:
            {
                if (logger.isWarnEnabled()) logger.warn("Skipping unsupported PRE_GA_UPDATE_ROWS_EVENT from: "
                        + logContext.getLogPosition());
                // ev = new Update_rows_log_event_old(buf, event_len,
                // description_event);
                break;
            }
            case LogEvent.PRE_GA_DELETE_ROWS_EVENT:
            {
                if (logger.isWarnEnabled()) logger.warn("Skipping unsupported PRE_GA_DELETE_ROWS_EVENT from: "
                        + logContext.getLogPosition());
                // ev = new Delete_rows_log_event_old(buf, event_len,
                // description_event);
                break;
            }
            case LogEvent.BEGIN_LOAD_QUERY_EVENT:
            {
                BeginLoadQueryLogEvent beginLoadQueryLogEvent = new BeginLoadQueryLogEvent(logHeader, logByteBlock, formatDescriptionLogEvent);
                /* updating index$originIndex in context */
                logPosition.position = logHeader.getLogPosition();
                return beginLoadQueryLogEvent;
            }
            case LogEvent.EXECUTE_LOAD_QUERY_EVENT:
            {
                ExecuteLoadQueryLogEvent executeLoadQueryLogEvent = new ExecuteLoadQueryLogEvent(logHeader, logByteBlock, formatDescriptionLogEvent);
                /* updating index$originIndex in context */
                logPosition.position = logHeader.getLogPosition();
                return executeLoadQueryLogEvent;
            }
            case LogEvent.INCIDENT_EVENT:
            {
                IncidentLogEvent incidentLogEvent = new IncidentLogEvent(logHeader, logByteBlock, formatDescriptionLogEvent);
                /* updating index$originIndex in context */
                logPosition.position = logHeader.getLogPosition();
                return incidentLogEvent;
            }
            case LogEvent.HEARTBEAT_LOG_EVENT:
            {
                HeartbeatLogEvent heartbeatLogEvent = new HeartbeatLogEvent(logHeader, logByteBlock, formatDescriptionLogEvent);
                /* updating index$originIndex in context */
                logPosition.position = logHeader.getLogPosition();
                return heartbeatLogEvent;
            }
            case LogEvent.IGNORABLE_LOG_EVENT:
            {
                IgnorableLogEvent ignorableLogEvent = new IgnorableLogEvent(logHeader, logByteBlock, formatDescriptionLogEvent);
                /* updating index$originIndex in context */
                logPosition.position = logHeader.getLogPosition();
                return ignorableLogEvent;
            }
            case LogEvent.ROWS_QUERY_LOG_EVENT:
            {
                RowsQueryLogEvent rowsQueryLogEvent = new RowsQueryLogEvent(logHeader, logByteBlock, formatDescriptionLogEvent);
                /* updating index$originIndex in context */
                logPosition.position = logHeader.getLogPosition();
                logHeader.putGtid(logContext.getGtidSet(), gtidLogEvent);
                return rowsQueryLogEvent;
            }
            case LogEvent.WRITE_ROWS_EVENT:
            {
                RowsLogEvent rowsLogEvent = new WriteRowsLogEvent(logHeader, logByteBlock, formatDescriptionLogEvent);
                /* updating index$originIndex in context */
                logPosition.position = logHeader.getLogPosition();
                rowsLogEvent.fillTableMapLogEvent(logContext);
                logHeader.putGtid(logContext.getGtidSet(), gtidLogEvent);
                return rowsLogEvent;
            }
            case LogEvent.UPDATE_ROWS_EVENT:
            {
                RowsLogEvent rowsLogEvent = new UpdateRowsLogEvent(logHeader, logByteBlock, formatDescriptionLogEvent);
                /* updating index$originIndex in context */
                logPosition.position = logHeader.getLogPosition();
                rowsLogEvent.fillTableMapLogEvent(logContext);
                logHeader.putGtid(logContext.getGtidSet(), gtidLogEvent);
                return rowsLogEvent;
            }
            case LogEvent.DELETE_ROWS_EVENT:
            {
                RowsLogEvent rowsLogEvent = new DeleteRowsLogEvent(logHeader, logByteBlock, formatDescriptionLogEvent);
                /* updating index$originIndex in context */
                logPosition.position = logHeader.getLogPosition();
                rowsLogEvent.fillTableMapLogEvent(logContext);
                logHeader.putGtid(logContext.getGtidSet(), gtidLogEvent);
                return rowsLogEvent;
            }
            case LogEvent.GTID_LOG_EVENT:
            case LogEvent.ANONYMOUS_GTID_LOG_EVENT:
            {
                GtidLogEvent $gtidLogEvent = new GtidLogEvent(logHeader, logByteBlock, formatDescriptionLogEvent);
                /* updating index$originIndex in context */
                logPosition.position = logHeader.getLogPosition();
                if (gtidSet != null)
                {
                    gtidSet.update($gtidLogEvent.getGtidString());
                    // update latest gtid
                    logHeader.putGtid(gtidSet, $gtidLogEvent);
                }
                // update current gtid event to context
                logContext.setGtidLogEvent($gtidLogEvent);
                return $gtidLogEvent;
            }
            case LogEvent.PREVIOUS_GTIDS_LOG_EVENT:
            {
                PreviousGtidsLogEvent previousGtidsLogEvent = new PreviousGtidsLogEvent(logHeader, logByteBlock, formatDescriptionLogEvent);
                /* updating index$originIndex in context */
                logPosition.position = logHeader.getLogPosition();
                return previousGtidsLogEvent;
            }
            case LogEvent.TRANSACTION_CONTEXT_EVENT:
            {
                TransactionContextLogEvent transactionContextLogEvent = new TransactionContextLogEvent(logHeader, logByteBlock, formatDescriptionLogEvent);
                /* updating index$originIndex in context */
                logPosition.position = logHeader.getLogPosition();
                return transactionContextLogEvent;
            }
            case LogEvent.VIEW_CHANGE_EVENT:
            {
                ViewChangeEvent viewChangeEvent = new ViewChangeEvent(logHeader, logByteBlock, formatDescriptionLogEvent);
                /* updating index$originIndex in context */
                logPosition.position = logHeader.getLogPosition();
                return viewChangeEvent;
            }
            case LogEvent.XA_PREPARE_LOG_EVENT:
            {
                XaPrepareLogEvent xaPrepareLogEvent = new XaPrepareLogEvent(logHeader, logByteBlock, formatDescriptionLogEvent);
                /* updating index$originIndex in context */
                logPosition.position = logHeader.getLogPosition();
                return xaPrepareLogEvent;
            }
            case LogEvent.ANNOTATE_ROWS_EVENT:
            {
                AnnotateRowsEvent annotateRowsEvent = new AnnotateRowsEvent(logHeader, logByteBlock, formatDescriptionLogEvent);
                /* updating index$originIndex in context */
                logPosition.position = logHeader.getLogPosition();
                logHeader.putGtid(logContext.getGtidSet(), gtidLogEvent);
                return annotateRowsEvent;
            }
            case LogEvent.BINLOG_CHECKPOINT_EVENT:
            {
                BinlogCheckPointLogEvent binlogCheckPointLogEvent = new BinlogCheckPointLogEvent(logHeader, logByteBlock, formatDescriptionLogEvent);
                /* updating index$originIndex in context */
                logPosition.position = logHeader.getLogPosition();
                return binlogCheckPointLogEvent;
            }
            case LogEvent.GTID_EVENT:
            {
                MariaGtidLogEvent mariaGtidLogEvent = new MariaGtidLogEvent(logHeader, logByteBlock, formatDescriptionLogEvent);
                /* updating index$originIndex in context */
                logPosition.position = logHeader.getLogPosition();
                return mariaGtidLogEvent;
            }
            case LogEvent.GTID_LIST_EVENT:
            {
                MariaGtidListLogEvent mariaGtidListLogEvent = new MariaGtidListLogEvent(logHeader, logByteBlock, formatDescriptionLogEvent);
                /* updating index$originIndex in context */
                logPosition.position = logHeader.getLogPosition();
                return mariaGtidListLogEvent;
            }
            case LogEvent.START_ENCRYPTION_EVENT:
            {
                StartEncryptionLogEvent startEncryptionLogEvent = new StartEncryptionLogEvent(logHeader, logByteBlock, formatDescriptionLogEvent);
                /* updating index$originIndex in context */
                logPosition.position = logHeader.getLogPosition();
                return startEncryptionLogEvent;
            }
            default:
                /*
                 * Create an object of Ignorable_log_event for unrecognized
                 * sub-class. So that SLAVE SQL THREAD will only update the
                 * index$originIndex and continue.
                 */
                if ((logByteBlock.getUint16(LogEvent.FLAGS_OFFSET) & LogEvent.LOG_EVENT_IGNORABLE_F) > 0)
                {
                    IgnorableLogEvent ignorableLogEvent = new IgnorableLogEvent(logHeader, logByteBlock, formatDescriptionLogEvent);
                    /* updating index$originIndex in context */
                    logPosition.position = logHeader.getLogPosition();
                    return ignorableLogEvent;
                } else
                {
                    if (logger.isWarnEnabled())
                    {
                        logger.warn("Skipping unrecognized binlog event " + LogEvent.getTypeName(logHeader.getType())
                                + " from: " + logContext.getLogPosition());
                    }
                }
        }

        /* updating index$originIndex in context */
        logPosition.position = logHeader.getLogPosition();
        /* Unknown or unsupported log event */
        return new UnknownLogEvent(logHeader);
    }

    public final void handle(final int fromIndex, final int toIndex)
    {
        handleSet.set(fromIndex, toIndex);
    }

    public final void handle(final int flagIndex)
    {
        handleSet.set(flagIndex);
    }

    /**
     * Decoding an event from binary-log bytes.
     *
     * @return <code>UknownLogEvent</code> if event type is unknown or skipped,
     * <code>null</code> if bytes is not including a full event.
     */
    public LogEvent decode(LogByteBlock logByteBlock, LogContext logContext) throws IOException
    {
        final int limit = logByteBlock.limit();

        if (limit >= FormatDescriptionLogEvent.LOG_EVENT_HEADER_LEN)
        {
            LogHeader logHeader = new LogHeader(logByteBlock, logContext.getFormatDescriptionLogEvent());

            final int eventLength = logHeader.getEventLength();
            if (limit >= eventLength)
            {
                LogEvent logEvent;

                /* Checking binary-log's headerPacket */
                if (handleSet.get(logHeader.getType()))
                {
                    logByteBlock.limit(eventLength);
                    try
                    {
                        /* Decoding binary-log to event */
                        logEvent = decode(logByteBlock, logHeader, logContext);
                        if (logEvent != null)
                        {
                            // set logFileName
                            logEvent.getLogHeader().setLogFileName(logContext.getLogPosition().getFileName());
                            logEvent.setSemival(logByteBlock.semival);
                        }
                    } catch (IOException e)
                    {
                        if (logger.isWarnEnabled())
                        {
                            logger.warn("Decoding " + LogEvent.getTypeName(logHeader.getType()) + " failed from: "
                                    + logContext.getLogPosition(), e);
                        }
                        throw e;
                    } finally
                    {
                        logByteBlock.limit(limit); /* Restore limit */
                    }
                } else
                {
                    /* Ignore unsupported binary-log. */
                    logEvent = new UnknownLogEvent(logHeader);
                }

                /* consume this binary-log. */
                logByteBlock.consume(eventLength);
                return logEvent;
            }
        }

        /* Rewind bytes's index$originIndex to 0. */
        logByteBlock.rewind();
        return null;
    }
}
